﻿using RdKafka;
using System;
using System.Collections.Generic;
using System.Text;
using System.Linq;
using System.Threading;

namespace AppSocketControl
{
    public class ReadKafakaConsumer
    { 
        static string brokerList = string.Empty;
        static List<string> topicList;
        static bool enableAutoCommit = false;
        static Topic topic;
        static EventConsumer eventConsumer = null;
        static int cDataServerCount = 0;
        public static List<GpsDataService> gpsDataServiceList;


        public static long cReadMessageCount = 0;
        public static long cLostMessageCount = 0;

        static ReadKafakaConsumer()
        {
            brokerList = ConfigurationManager.AppSettings["Kafka:BrokerList"];
            topicList  = new List<string>() { ConfigurationManager.AppSettings["Kafka:TopicName"] }; ;
            enableAutoCommit = bool.Parse( ConfigurationManager.AppSettings["Kafka:EnableAutoCommit"]);

            Config config = new Config()
            {
                GroupId ="GPS-MAIN",
                EnableAutoCommit = enableAutoCommit,
                StatisticsInterval = TimeSpan.FromSeconds(60),
            };

            eventConsumer = new EventConsumer(config, brokerList);
            var a = eventConsumer.GetWatermarkOffsets(new TopicPartition("GPS-MAIN",0));


        }

        public static void  Start()
        { 
            eventConsumer.OnMessage += EventConsumer_OnMessage;

            eventConsumer.OnEndReached += EventConsumer_OnEndReached;
            eventConsumer.OnConsumerError += EventConsumer_OnConsumerError;
            eventConsumer.OnError += EventConsumer_OnError;
            eventConsumer.OnPartitionsAssigned += EventConsumer_OnPartitionsAssigned;
            eventConsumer.OnPartitionsRevoked += EventConsumer_OnPartitionsRevoked;
            //eventConsumer.OnStatistics += EventConsumer_OnStatistics;

            if (enableAutoCommit)
            {
                eventConsumer.OnOffsetCommit += EventConsumer_OnOffsetCommit;
            }
            eventConsumer.Subscribe(topicList);
            eventConsumer.Start();

            //TopicPartitionOffset topicPartitionOffset = new TopicPartitionOffset("GPS-MAIN", 0, 0);
            //List<TopicPartitionOffset> list = new List<TopicPartitionOffset>();
            //list.Add(topicPartitionOffset);
            //eventConsumer.Assign(list);
        }

        private static void stop()
        {
            eventConsumer.Stop();
            Thread.Sleep(1000);
        }

        #region   消息处理
        private static void EventConsumer_OnOffsetCommit(object sender, Consumer.OffsetCommitArgs commit)
        {
            string str = string.Empty;
            if (commit.Error != ErrorCode.NO_ERROR)
            {
                str = $"  Failed to commit offsets:{commit.Error}";
                WriteLine(str);

            }
            str = $"  Sussessfully committed offsets:[{string.Join(", ", commit.Offsets)}]";
            WriteLine(str);
        }

        private static void EventConsumer_OnStatistics(object sender, string json)
        {
            string str = string.Empty;
            str = $"  Statistics: {json}";
            WriteLine(str);
        }

        private static void EventConsumer_OnPartitionsRevoked(object sender, List<TopicPartitionOffset> partitions)
        {
            string str = string.Empty;
            str = $"  Revoked partitions: [{string.Join(", ", partitions)}]";
            WriteLine(str);
            eventConsumer.Unassign();
        }

        private static void EventConsumer_OnPartitionsAssigned(object sender, List<TopicPartitionOffset> partitions)
        {
            string str = string.Empty;
            str = $"  Assigned partitions : [{string.Join(", ", partitions)}]";
            WriteLine(str);
            eventConsumer.Assign(partitions);
        }

        private static void EventConsumer_OnError(object sender, Handle.ErrorArgs e)
        {
            string str = string.Empty;
            str = $"  Error:{e.ErrorCode}  {e.Reason}";
            WriteLine(str);
        }

        private static void EventConsumer_OnEndReached(object sender, TopicPartitionOffset e)
        {
            string str = string.Empty;
            str = $"  Reached end of topic: { e.Topic} partition: {e.Partition}, next message will be at offset {e.Offset} ";
            WriteLine(str);
        }

        private static void EventConsumer_OnConsumerError(object sender, ErrorCode e)
        {
            string str = ($"  Consumer Error:{e}");
            WriteLine(str);
        }
        #endregion

        /// <summary>
        /// 读取一条kafka数据
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="msg"></param>
        private static void EventConsumer_OnMessage(object sender, Message msg)
        { 
            string text = Encoding.UTF8.GetString(msg.Payload, 0, msg.Payload.Length);
         
            string str = $"  Topic:={msg.Topic}    Partition:={msg.Partition}   Offset:={msg.Offset} " ;
            text += $"$Offset:={msg.Offset}";
            WriteLine(str);
            WriteLine(text);

            if(gpsDataServiceList != null && gpsDataServiceList.Count>0)
            {
                if (gpsDataServiceList.Count(s=>s.IsIdel)>0)
                {
                    var svr = gpsDataServiceList.First(s => s.IsIdel);
                    WriteLine($"  IsIdel Thread Count := {gpsDataServiceList.Count(s => s.IsIdel)} , CurrentSeleteThreadNo:= {svr.ThreadNumber}");
                    svr.SetMsg(text);
                    cReadMessageCount++;
                    WriteLine($"  cReadMessageCount:= {cReadMessageCount} , cLostMessageCount:= {cLostMessageCount}");
                    return;
                }
                else
                {
                    cLostMessageCount++;
                    WriteLine($"  cLostMessageCount:= {cLostMessageCount}  ,  cReadMessageCount:= {cReadMessageCount} ");
                }
            }
            else
            {
                cLostMessageCount++;
                WriteLine($"  cLostMessageCount:= {cLostMessageCount}  ,  cReadMessageCount:= {cReadMessageCount} ");
            }

            if (!enableAutoCommit && msg.Offset % 10 == 0)
            {
                str = ("  Commiting offset");
                WriteLine(str);
                eventConsumer.Commit(msg).GetAwaiter().GetResult();
                str = ("  Commited offset");
                WriteLine(str);
            }
        }

        private static void WriteLine(string msg)
        {
            Console.WriteLine(msg); 
        } 
          
    }
     
}
